#!/usr/bin/python3
# -*- coding: utf-8 -*-
# ======================================#
# @Author   : duanqizhong
# @time     : 2022-8-6 23:26
# @file     : cls
# @Software : PyCharm
# ======================================#
import ftplib
import json
import os
from configparser import ConfigParser
from ftplib import FTP_PORT

import cx_Oracle
import pymssql
import pymysql as pymysql
import redis
import requests
from kafka import KafkaClient, KafkaAdminClient, KafkaConsumer
from openpyxl import load_workbook, Workbook
from rediscluster import RedisCluster

from base.log.logger import log
from settings import CONNECT_TIMEOUT
from settings import FTP_ENCODING, FTP_SERVER, FTP_BUFFER_SIZE


class Mysql:
    def __init__(self, conn):
        self.conn = conn
        self.db = None

    def connect(self, database):
        """连接数据库"""
        self.db = pymysql.connect(host=self.conn["host"],
                                  port=self.conn["port"],
                                  user=self.conn["username"],
                                  passwd=self.conn["password"],
                                  database=database,
                                  use_unicode=True,
                                  charset='utf8')
        return self

    def commit(self):
        if self.db is not None:
            self.db.commit()

    def close(self):
        if self.db is not None:
            self.db.close()
            self.db = None

    def execute(self, sql):
        cursor = self.db.cursor()
        try:
            cursor.execute(sql)
            cursor.close()
        except Exception as e:
            self.db.rollback()
            log.error(f"{self.conn}# {e}")
            return False

        return True

    def fetchone(self, sql, key_lower=False):
        cursor = self.db.cursor()
        cursor.execute(sql)
        if key_lower:
            columns = [col[0].lower() for col in cursor.description]
        else:
            columns = [col[0] for col in cursor.description]
        record = cursor.fetchone()
        cursor.close()
        if record:
            return dict(zip(columns, record))
        return None

    def fetchall(self, sql, key_lower=False):
        cursor = self.db.cursor()
        cursor.execute(sql)
        if key_lower:
            columns = [col[0].lower() for col in cursor.description]
        else:
            columns = [col[0] for col in cursor.description]
        all_result = [dict(zip(columns, row)) for row in cursor.fetchall()]
        cursor.close()
        return all_result


class Oracle:
    def __init__(self, conn):
        self.conn = conn
        self.db = None

    def connect(self, database):
        self.db = cx_Oracle.connect(self.conn["username"], self.conn["password"],
                                    f"{self.conn['host']}:{self.conn['port']}/{database}", encoding="UTF-8")
        return self

    def check_connect(self, database):
        try:
            self.db = cx_Oracle.connect(host=self.conn["host"],
                                        port=self.conn["port"],
                                        user=self.conn["username"],
                                        passwd=self.conn["password"],
                                        database=database)
            self.close()
            return True
        except Exception as e:
            self.db = None
            log.error(e)
            return False

    def check_table(self, database, table):
        sql = "show tables"
        self.connect(database)
        results = self.fetchall(sql)
        for result in results:
            if table in result.values():
                self.close()
                return True
        self.close()
        return False

    def get_tables(self, database):
        sql = "show tables"
        self.connect(database)
        results = self.fetchall(sql)
        self.close()
        tables = []
        for result in results:
            for key, value in result.items():
                tables.append(value)
        return tables

    def commit(self):
        if self.db is not None:
            self.db.commit()

    def close(self):
        if self.db is not None:
            self.db.close()
            self.db = None

    def execute(self, sql):
        cursor = self.db.cursor()
        try:
            cursor.execute(sql)
        except Exception as e:
            self.db.rollback()
            print(e)
            return False
        return True

    def fetchone_crude(self, sql):
        cursor = self.db.cursor()
        cursor.execute(sql)
        return cursor.fetchone()

    def fetchall_crude(self, sql):
        cursor = self.db.cursor()
        cursor.execute(sql)
        return cursor.fetchall()

    def fetchone(self, sql):
        cursor = self.db.cursor()
        cursor.execute(sql)
        columns = [col[0] for col in cursor.description]
        record = cursor.fetchone()
        if record:
            return dict(zip(columns, record))
        return None

    def fetchall(self, sql):
        cursor = self.db.cursor()
        cursor.execute(sql)
        columns = [col[0] for col in cursor.description]
        return [dict(zip(columns, row)) for row in cursor.fetchall()]


class SQLServer:
    def __init__(self, conn):
        self.conn = conn
        self.db = None

    def connect(self, database):
        self.db = pymssql.connect(server=self.conn["host"],
                                  user=self.conn["username"],
                                  password=self.conn["password"],
                                  database=database)
        return self

    def check_connect(self, database):
        try:
            self.db = pymssql.connect(host=self.conn["host"],
                                      user=self.conn["username"],
                                      password=self.conn["password"],
                                      database=database)
            self.close()
            return True
        except Exception as e:
            self.db = None
            log.error(e)
            return False

    def check_table(self, database, table):
        sql = "show tables"
        self.connect(database)
        results = self.fetchall(sql)
        for result in results:
            if table in result.values():
                self.close()
                return True
        self.close()
        return False

    def get_tables(self, database):
        sql = "show tables"
        self.connect(database)
        results = self.fetchall(sql)
        self.close()
        tables = []
        for result in results:
            for key, value in result.items():
                tables.append(value)
        return tables

    def commit(self):
        if self.db is not None:
            self.db.commit()

    def close(self):
        if self.db is not None:
            self.db.close()
            self.db = None

    def execute(self, sql):
        cursor = self.db.cursor()
        try:
            cursor.execute(sql)
        except Exception as e:
            self.db.rollback()
            print(e)
            return False
        return True

    def fetchone_crude(self, sql):
        cursor = self.db.cursor()
        cursor.execute(sql)
        return cursor.fetchone()

    def fetchall_crude(self, sql):
        cursor = self.db.cursor()
        cursor.execute(sql)
        return cursor.fetchall()

    def fetchone(self, sql):
        cursor = self.db.cursor()
        cursor.execute(sql)
        columns = [col[0] for col in cursor.description]
        record = cursor.fetchone()
        if record:
            return dict(zip(columns, record))
        return None

    def fetchall(self, sql):
        cursor = self.db.cursor()
        cursor.execute(sql)
        columns = [col[0] for col in cursor.description]
        return [dict(zip(columns, row)) for row in cursor.fetchall()]


class KafkaProduct:
    def __init__(self, hosts, topic):
        self.client = KafkaClient(hosts=hosts)

    def set_topics(self, topic):
        self.__topic = self.client.set_topics(topic)

    def get_topics(self):
        return self.client._topics

    def producer(self):
        with self.get_topics().get_producer(delivery_reports=True) as producer:
            next_data = ''
            while True:
                if next_data:
                    producer.produce(bytes(next_data))
                next_data = yield True

    def send_data(self, data):
        c = self.producer()
        next(c)
        for msg in data:
            c.send(msg)


class KafkaConsume:
    def __init__(self, hosts, group_id):
        self.client = KafkaConsumer(bootstrap_servers=hosts, group_id=group_id)
        self.all_topics = self.get_topics()
        self.topics = None

    def set_topic(self, topic):
        if topic in self.all_topics:
            self.topics = topic

    def get_topics(self):
        self.all_topics = self.client.topics()
        return self.all_topics

    def get_topic(self):
        return self.topics

    def beginning_offsets(self, partitions):
        self.client.beginning_offsets(partitions)


class KafkaAdmin:
    def __init__(self, hosts):
        self.admin = KafkaAdminClient(bootstrap_servers=hosts)

    def get_topics(self):
        return self.admin.list_topics()

    def get_topic(self):
        pass

    def is_exists(self, topic):
        pass

    def describe_group(self, group_ids):
        return self.admin.describe_consumer_groups(group_ids=group_ids)

    def list_consumers(self):
        return self.admin.list_consumer_groups()

    def list_consumer_offsets(self, group_id="consumer"):
        return self.admin.list_consumer_group_offsets(group_id=group_id)


class Ftp:
    def __init__(self, host=FTP_SERVER, port=FTP_PORT):
        self.host = host
        self.port = port
        self.op = ftplib.FTP()
        # 重新设置下编码方式
        self.op.encoding = FTP_ENCODING

    def login(self, username, password):
        login_status = False
        try:
            # 0主动模式 1被动方式
            self.op.set_pasv(True)
            # 打开调试级别2，显示详细信息
            # self.op.set_debuglevel(2)
            self.op.connect(self.host, self.port)
            self.op.login(username, password)
            login_status = True
        except Exception as e:
            log.error(f"FTP连接或登录失败：{e}")
        return login_status

    def close(self):
        self.op.quit()

    def upload_file(self, local_file, remote_file, buf_size=FTP_BUFFER_SIZE):
        """
        已存在的文件无法覆盖，会报错：需要先删除旧文件然后上传新的，原则上不允许删除旧文件，上传需要添加日期
        :param buf_size:
        :param local_file:      本地文件
        :param remote_file:     远程文件
        :return:
        """
        upload_status = False
        if not os.path.exists(local_file):
            log.error(f"FTP待上传文件路径不存在：{local_file}")
            return upload_status

        # 大小名字一样跳过
        if self.is_same_size(local_file, remote_file):
            log.warn(f"FTP远程已存在相同大小文件: {remote_file}")
            return True

        fp = open(local_file, 'rb')
        try:
            self.op.storbinary(f"STOR {remote_file}", fp, buf_size)
            upload_status = True
        except Exception as e:
            log.error(f"FTP文件上传失败：{local_file} {e}")
            return upload_status
        finally:
            fp.close()
        log.info(f"FTP文件上传成功：{remote_file}")
        return upload_status

    def download_file(self, local_file, remote_file, buf_size=FTP_BUFFER_SIZE):
        """
        :param buf_size:
        :param local_file:      本地文件
        :param remote_file:     远程文件
        :param bufsize:
        :return:
        """
        download_status = False
        if self.is_same_size(local_file, remote_file):
            log.warn(f"FTP未下载文件，本地已存在：{local_file}")
            return download_status
        fp = open(local_file, 'wb')
        try:
            self.op.retrbinary(f"RETR {remote_file}", fp.write, buf_size)
            download_status = True
        except Exception as e:
            log.error(f"FTP文件下载失败：{remote_file} {e}")
            return download_status
        finally:
            fp.close()

        log.info(f"FTP下载文件成功：{local_file}")
        return download_status

    def download_file_tree(self, local_path, remote_path):
        """
        从远程目录下载多个文件到本地
        :param local_path:          本地路径
        :param remote_path:         远程路径
        :return:
        """
        download_status = False
        try:
            # 切换工作目录
            self.op.cwd(remote_path)
        except Exception as e:
            log.error(f"FTP远程目录不存在：{remote_path} {e}")
            return download_status

        # 先查看一下指定的文件路径是否存在，如果本地不存在，创建对应的文件夹
        self.check_local_dir_or_create(local_path)

        file_list = self.op.nlst()

        for file_name in file_list:
            src = os.path.join(local_path, file_name)
            # 远程文件是文件夹 or 文件
            if self.check_ftp_dir(file_name):
                # 是目录的话，查看本地存不存在目录，不存在创建，递归ftp获取内层文件
                self.check_local_dir_or_create(src)
                # 递归: src是local_dir file_name是remote_dir
                self.download_file_tree(src, file_name)

            else:
                # 不是目录的话，开始下载:src是local_file file_name是remote_file
                self.download_file(src, file_name)

        self.op.cwd("..")
        return True

    def check_local_dir_or_create(self, local_dir):
        """检查本地文件存在不存在，不存在创建对应的文件夹"""
        if not os.path.isdir(local_dir):
            os.makedirs(local_dir)

    def check_ftp_dir(self, dir_path):
        """探测dir_path是不是一个文件夹"""
        try:
            self.op.cwd(dir_path)
            self.op.cwd("..")
        except Exception as e:
            log.error(str(e))
            return False
        return True

    def is_same_size(self, local_file, remote_file):
        """判断远程文件和本地文件大小是否一致"""
        try:
            remote_file_size = self.op.size(remote_file)
            local_file_size = os.path.getsize(local_file)
        except Exception as e:
            log.error(str(e))
            return False

        return remote_file_size == local_file_size

    def delete_remote(self, remote_file):
        try:
            self.op.delete(remote_file)
        except Exception as e:
            log.error(f"FTP删除失败：文件不存在或没有权限删除 {remote_file} {e}")
            return False
        log.info(f"FTP已删除：{remote_file}")
        return True


class Redis:
    def __init__(self, redis_type="StrictRedis", host="127.0.0.1", port=6379, db=0):
        if redis_type == "StrictRedis":
            self.r = redis.StrictRedis(host=host, port=port, db=db, socket_connect_timeout=1)
        elif redis_type == "Redis":
            self.r = redis.Redis(host=host, port=port, db=db, socket_connect_timeout=1)
        elif redis_type == "ReidsPool":
            pool = redis.ConnectionPool(host=host, port=port, db=db, socket_connect_timeout=1)
            self.r = redis.Redis(connection_pool=pool)
        else:
            self.r = redis.StrictRedis(host, port, db)

    def set_json(self, key, value, expire=600):
        """写入dict"""
        assert isinstance(value, dict), "dict value is required"
        bytes_value = bytes(json.dumps(value), encoding="utf-8")
        return self.r.set(key, bytes_value, ex=expire)

    def get_json(self, key):
        """读取json"""
        value = self.r.get(key)
        json_value = json.loads(value.decode("utf-8")) if value else value
        return json_value

    def set(self, key, value, expire=600):
        """写入键值对"""
        return self.r.set(key, value, ex=expire)

    def get(self, key):
        """读取键值对内容"""
        value = self.r.get(key)
        return value.decode("utf-8") if value else value

    def delete(self, *names):
        """删除一个或多个"""
        return self.r.delete(*names)

    def hset(self, name, key, value):
        """写入hash表"""
        return self.r.hset(name, key, value)

    def hmset(self, key, *value):
        """获取指定hash表所有给定字段的值"""
        value = self.r.hmset(key, *value)
        return value

    def hget(self, name, key):
        """读取指定hash表值"""
        value = self.r.hget(name, key)
        return value.decode("utf-8") if value else value

    def hgetall(self, name):
        """获取指定hash表所有值"""
        return self.r.hgetall(name)

    def hdel(self, name, key):
        """删除指定hash表键值"""
        return self.r.hdel(name, key)

    def keys(self):
        """获取所有的keys"""
        return self.r.keys()

    def expire(self, name, expire=600):
        """设置过期时间"""
        return self.r.expire(name, expire)

    def flushdb(self):
        return self.r.flushdb()


class MyRedisCluster:
    """host格式[{"host":"127.0.0.1","port":6379},{"host":"127.0.0.1","port":6380}]"""

    def __init__(self, hosts, username=None, password=None):
        self.startup_nodes = hosts
        self.username = username
        self.password = password
        self.rc = self.connect()

    def connect(self):
        try:
            redisconn = RedisCluster(startup_nodes=self.startup_nodes, decode_responses=True)
        except Exception as e:
            log.error(f"redis cluster connect failed: {e}")
            redisconn = None
        return redisconn

    def get(self, key):
        return self.rc.get(key)

    def keys(self, pattern=None):
        if pattern:
            return self.rc.keys(pattern)
        return self.rc.keys()

    def set(self, key, value, ex=900):
        return self.rc.set(name=key, value=value, ex=ex)

    def delete(self, key):
        return self.rc.delete(key)

    def flushdb(self):
        return self.rc.flushdb()


class Ding:
    def __init__(self, token):
        self.url = f"https://oapi.dingtalk.com/robot/send?access_token={token}"
        self.headers = {"content-Type": "application/json;charset=UTF-8"}

    def send_text(self, text):
        textMsg = {
            "msgtype": "text",
            "text": {"content": text}
        }
        try:
            requests.post(self.url, json=textMsg, headers=self.headers, timeout=CONNECT_TIMEOUT)
        except Exception as e:
            log.error(f"send_dingding发送失败：{e}")
            pass

    def send_link(self, title, text, link, token):
        # token='f380f3b2791bec3dc584544703432aa3b4ce81af8b9bf82df936ca5ee10bf6ef'
        testMsg = {
            "msgtype": "link",
            "link": {
                "text": text,
                "title": title,
                "picUrl": "",
                "messageUrl": link
            }
        }
        try:
            send_msg = json.dumps(testMsg).encode("utf-8")
            requests.post(url=self.url, data=send_msg, headers=self.headers, timeout=5)
        except Exception as e:
            log.error(f"send_dingding_link发送失败：{e}")
            pass


class FeiShu:
    def __init__(self, token):
        self.url = f"https://open.feishu.cn/open-apis/bot/hook/{token}"
        self.headers = {"content-Type": "application/json;charset=UTF-8"}

    def send_text(self, title, text):
        textMsg = {
            "title": title,
            "text": text
        }
        try:
            requests.post(self.url, json=textMsg, headers=self.headers, timeout=CONNECT_TIMEOUT)
        except Exception as e:
            log.error(f"send_feishu发送失败：{e}")
            pass


class QYWX:
    def __init__(self, token):
        self.url = f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={token}"
        self.headers = {"content-Type": "application/json;charset=UTF-8"}
        self.timeout = 3

    def send_text(self, title, text):
        textMsg = {
            "msgtype": "text",
            "text": {"content": f"{title}:{text}"}
        }
        try:
            requests.post(self.url, json=textMsg, headers=self.headers, timeout=self.timeout)
        except Exception as e:
            log.error(f"send_qiyewx发送失败：{e}")
            return False
        return True


class XLS:
    def __init__(self, path):
        self.path = path

    def read(self, sheet_name="Sheet1"):
        workbook = load_workbook(self.path, read_only=True)
        worksheet = workbook[sheet_name]
        data = []
        for row in worksheet.rows:
            row_info = []
            if not row[0].value:
                continue
            for cell in row:
                if cell.value == None:
                    break
                row_info.append(cell.value)

            data.append(row_info.copy())
        return data

    def write(self, data, sheet_name="Sheet1"):
        assert self.path.split('.')[-1].lower() == 'xlsx', "filename format: filename.xlsx"
        workbook = Workbook(write_only=True)
        worksheet = workbook.create_sheet(title=sheet_name)
        for row in data:
            worksheet.append(row)
        workbook.save(self.path)
        return True


class ParseConf:
    """解析conf"""

    def __init__(self, path, encoding="utf-8"):
        if not os.path.exists(path):
            raise Exception(f"path={path} not found!")
        self.path = path
        self.parse = ConfigParser()
        self.parse.read(path, encoding=encoding)
        # self.parse.read(path)

    def get_section_options(self, section):
        """获取配置文件内section的options"""
        return self.parse.options(section)

    def get_sections(self):
        """获取配置文件内sections"""
        return self.parse.sections()

    def check_section(self, section):
        """检查指定的section是否存在"""
        if section in self.get_sections():
            return True
        else:
            return None

    def check_option(self, section, option):
        """检查是否存在option"""
        if not self.check_section(section):
            return False
        if option in self.get_section_options(section):
            return True
        else:
            return False

    def get_section_info(self, section):
        """
        获取section的key=value
        default section的值会被获取到，被作为默认值
        如果指定的section不存在，返回{}；否则返回key,value
        """
        section_info = {}
        if self.check_section('default'):
            default_options = self.get_section_options('default')
            for option in default_options:
                section_info[option] = self.parse.get('default', option)
        if section == 'default':
            return section_info
        if self.check_section(section):
            section_options = self.get_section_options(section)
            for option in section_options:
                section_info[option] = self.parse.get(section, option)
        else:
            return {}
        return section_info

    def get_sections_info(self):
        """获取所有sections info，section:{key=value}"""
        sections = self.get_sections()
        sections_info = {}
        for section in sections:
            sections_info[section] = self.get_section_info(section)
        return sections_info

    def set_section_info(self, section):
        if not self.check_section(section):
            self.parse.add_section(section)
            return True
        return False

    def add_section_info(self, section, key, value):
        if not self.check_section(section):
            self.set_section_info(section)
        self.parse.set(section, key, value)
        return True

    def writeback(self, encoding='utf-8'):
        self.parse.write(open(self.path, 'w', encoding=encoding))


class Cartesian:
    """
    数组 (笛卡尔积): 生成组合测试用例（数组不能为空，否则生成0个组合）
    示例： a=[1,2,3];b="";c=[4,5];d=["a","b","c"]; co=[a,b,c,d]
    result=[[1, '', 4, 'a'], [1, '', 4, 'b'], [1, '', 4, 'c'], [1, '', 5, 'a'], [1, '', 5, 'b'], [1, '', 5, 'c'],
            [2, '', 4, 'a'], [2, '', 4, 'b'], [2, '', 4, 'c'], [2, '', 5, 'a'], [2, '', 5, 'b'],
            [2, '', 5, 'c'], [3, '', 4, 'a'], [3, '', 4, 'b'], [3, '', 4, 'c'], [3, '', 5, 'a'],
            [3, '', 5, 'b'], [3, '', 5, 'c']]
    """

    def __init__(self, datagroup):
        assert datagroup, "不能为空"
        self.datagroup = datagroup
        # 二维数组下标值（从下向上）
        self.counterIndex = len(datagroup) - 1
        # 每次输出列变化的下标值数组(初始化为0)
        self.counter = [0 for i in range(0, len(self.datagroup))]

    def countlength(self):
        """计算数组长度；即可以组成多少个组合"""
        i = 0
        length = 1
        while i < len(self.datagroup):
            if isinstance(self.datagroup[i], list) or isinstance(self.datagroup[i], tuple):
                length *= len(self.datagroup[i])
            else:
                length *= 1
            i += 1
        return length

    def __handle(self):
        """核心：递归处理下标，通过改变矢量下标，组合成所有可以组合的情况；每次生成一个记录列变化的下标数组"""
        if isinstance(self.datagroup[self.counterIndex], list) or isinstance(self.datagroup[self.counterIndex], tuple):
            # 从最后一个数组开始变化下标，下标依次递增
            self.counter[self.counterIndex] += 1
            # 下标是否超过数组长度；超过长度，遍历结束，移动到上一个数组
            if self.counter[self.counterIndex] >= len(self.datagroup[self.counterIndex]):
                # 重置当前行的下标为0
                self.counter[self.counterIndex] = 0
                # 向上移动到上一个数组
                self.counterIndex -= 1
                # 是否是第一个数组，不是的话，进行递归，直到算出一个矢量数组
                if self.counterIndex >= 0:
                    self.__handle()
                # 重置标记
                self.counterIndex = len(self.datagroup) - 1
        else:
            # 非list，当前行下标置为0，然后上移数组递归
            self.counter[self.counterIndex] = 0
            self.counterIndex -= 1
            if self.counterIndex >= 0:
                self.__handle()
        return self.counter

    def assemble(self):
        """排列组合输出"""
        # 计算总共有多少个组合
        lists = []
        length = self.countlength()
        i = 0
        while i < length:
            attrlist = []
            # 维度
            j = 0
            while j < len(self.datagroup):
                if isinstance(self.datagroup[j], list) or isinstance(self.datagroup[j], tuple):
                    attrlist.append(self.datagroup[j][self.counter[j]])
                else:
                    attrlist.append(self.datagroup[j])
                j += 1

            lists.append(attrlist)
            self.__handle()
            i += 1
        return lists
